package com.gitee.melin.bee.core.fs.impl;

import com.gitee.melin.bee.core.fs.FileSystem;
import com.gitee.melin.bee.core.fs.FsConnectionInfo;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.Bucket;
import software.amazon.awssdk.services.s3.model.ListBucketsResponse;
import software.amazon.awssdk.services.sts.StsClient;
import software.amazon.awssdk.services.sts.model.GetCallerIdentityResponse;

public class AwsS3FileSystem implements FileSystem {
    private static final Logger LOGGER = LoggerFactory.getLogger(AwsS3FileSystem.class);

    @Override
    public String testConnection(FsConnectionInfo connectionInfo) {
        AwsCredentials awsCredentials =
                AwsBasicCredentials.create(connectionInfo.getAccessKeyId(), connectionInfo.getSecretAccessKey());
        StaticCredentialsProvider provider = StaticCredentialsProvider.create(awsCredentials);

        try (StsClient stsClient = StsClient.builder()
                .credentialsProvider(provider)
                .region(Region.of(connectionInfo.getRegion()))
                .httpClient(ApacheHttpClient.builder().build())
                .build()) {
            GetCallerIdentityResponse response = stsClient.getCallerIdentity();
            LOGGER.info("s3 client connection success: " + response);

            String bucket = connectionInfo.getBucket();
            if (StringUtils.isNotBlank(bucket)) {
                List<String> buckets = this.queryBuckets(connectionInfo.getRegion(), provider);
                if (!buckets.contains(bucket)) {
                    throw new IllegalArgumentException("bucket not exist: " + bucket);
                }
            }

            return response.toString();
        }
    }

    @Override
    public List<String> queryBuckets(FsConnectionInfo connectionInfo) {
        AwsCredentials awsCredentials =
                AwsBasicCredentials.create(connectionInfo.getAccessKeyId(), connectionInfo.getSecretAccessKey());
        StaticCredentialsProvider provider = StaticCredentialsProvider.create(awsCredentials);

        return queryBuckets(connectionInfo.getRegion(), provider);
    }

    private List<String> queryBuckets(String region, StaticCredentialsProvider provider) {
        try (S3Client s3Client = S3Client.builder()
                .httpClient(ApacheHttpClient.builder().build())
                .region(Region.of(region))
                .credentialsProvider(provider)
                .build()) {

            ListBucketsResponse bucketsResponse = s3Client.listBuckets();
            return bucketsResponse.buckets().stream().map(Bucket::name).collect(Collectors.toList());
        }
    }
}
